package com.saic.produce;

import java.io.IOException;
import java.io.InputStream;
import java.util.List;
import java.util.Map;
import java.util.Properties;

import org.json.simple.JSONObject;

import com.saic.utils.OracleUtils;

import kafka.javaapi.producer.Producer;
import kafka.producer.KeyedMessage;
import kafka.producer.ProducerConfig;

public class OracleProduce {
	
	public static String topic ;
	public static Producer<String, String> producer;
	
	public static void main(String[] args){
		OracleProduce.sendtoKafka(OracleUtils.getDate("T_ETL_JT_HPL_D_EXP_VIEW", "VIEW_ID"));
		System.out.println("send msg success! ");
	}

	static {
		InputStream in = OracleProduce.class.getClassLoader().getResourceAsStream("kafka.properties");
		Properties property = new Properties();
		try {
			property.load(in);
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		topic = property.getProperty("kafka.topic");
		String brokerlist = property.getProperty("kafka.brokerlist");
		System.out.println("=====================================================");
		System.out.println(topic+","+brokerlist);
		Properties properties = new Properties();
		properties.put("metadata.broker.list", brokerlist);
		properties.put("serializer.class", "kafka.serializer.StringEncoder");
		ProducerConfig producerConfig = new ProducerConfig(properties);
		producer = new kafka.javaapi.producer.Producer<String, String>(producerConfig);
		try {
			in.close();
		} catch (IOException e) {
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
	}

	/**
	 * 把JSON字符串存入内存队列
	 * 
	 * @param JSONString
	 * @return
	 */
	public static  boolean sendtoKafka(List<Map<String,String>> listMap) {
		try {
			for(Map<String,String> map : listMap){
			producer.send(new KeyedMessage<String, String>(topic, JSONObject.toJSONString(map)));
			}
		} catch (Exception e) {
			// TODO: handle exception
			e.printStackTrace();
			return false;
		}
		return true;
	}


}
